package com.imooc.bigdata.chapter08_my.business

import com.imooc.bigdata.chapter08.utils.KuduUtils
import com.imooc.bigdata.chapter08_my.utils.{SQLUtils, SchemaUtils}
import org.apache.kudu.ColumnSchema.ColumnSchemaBuilder
import org.apache.kudu.{Schema, Type}
import org.apache.spark.sql.{DataFrame, SparkSession}

//统计：按照provincename，cityname的分组统计
//
//    扩展：求每个省份下每个城市数量最多的TopN   这是一个典型的To
object ProvinceCityStatApp {

  def main(args: Array[String]): Unit = {

    val spark: SparkSession = SparkSession.builder()
      .appName("ProvinceCityStatApp")
      .master("local[2]")
      .getOrCreate()

    val SOURCE_TABLE_NAME = "ods"
    val MASTER="spark000"
    //从kudu 的ODS表中读取数据，然后按照省份和城市分组统计
    var odsDF: DataFrame = spark.read.format("org.apache.kudu.spark.kudu")
      .option("kudu.table", SOURCE_TABLE_NAME)
      .option("kudu.master", MASTER)
      .load()

   // 扩展：求每个省份下每个城市数量最多的TopN   这是一个典型的TopN


    //写sql得到结果
    odsDF.createOrReplaceTempView("ods")
    var resultFD: DataFrame = spark.sql(SQLUtils.PROVINCE_CITY_SQL)



    //写回kudu
    val SINK_TABLE_NAME ="province_city_stat"
    val partitionId = "provincename"
    var schema: Schema = SchemaUtils.ProvinceCitySchema

    KuduUtils.sink(resultFD,SINK_TABLE_NAME,MASTER,schema,partitionId)
    spark.close()
  }

}
